package com.ayu.flink.datasteamapibase;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @Author: 徐林玉//作者及
 * @Date: 2022/3/23//完成日期
 * @Description: //
 * @Version: v0.0.1 // 版本信息
 * @Function List: // 主要函数及其功能
 * @Others: // 其它内容的说明
 * @History: // 历史修改记录
 **/
public class TransformReduceTest
{

	public static void main(String[] args) throws Exception
	{
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

		env.setParallelism(1);

		//获取数据源
		DataStreamSource<Event> sourceStream = env.addSource(new ClickSource());

		sourceStream.map(data -> data).keyBy(data->data.user).reduce((v1,v2)->v1.timestamp>v2.timestamp?v1:v2).print("test");



		//统计每个用户的点击量
		SingleOutputStreamOperator<Tuple2<String, Long>> clickStream = sourceStream.map(data -> Tuple2.of(data.user, 1L)).returns(Types.TUPLE(Types.STRING,Types.LONG)).keyBy(data -> data.f0).reduce((v1, v2) -> Tuple2.of(v1.f0, v1.f1 + v2.f1));

//		clickStream.print("user:");


		//统计最活跃的用户
		SingleOutputStreamOperator<Tuple2<String, Long>> reduce = clickStream.keyBy(data -> "data").reduce((v1, v2) -> v1.f1 > v2.f1 ? v1 : v2);


//		reduce.print("all:");

		env.execute();

	}
}
